package PassengerVehicle

import java.util.Properties

import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 载客率
  */
object LoadFactors {

  def main(args: Array[String]): Unit = {
    //创建sparkSession
    val spark: SparkSession = SparkSession.builder().appName("Vehicle").master("local[4]").config("spark.debug.maxToStringFields", "1000").getOrCreate()
    //创建properties对象 设置连接mysql的信息
    val prop: Properties = new Properties()
    prop.setProperty("user", "root")
    prop.setProperty("password", "123456")

    //读取mysql数据
    val mysqlDF: DataFrame = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/localsql", "zhjg_jsy_ticket_smz_curmonth", prop)
    val vehicleDF: DataFrame = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/localsql", "zhjg_zy_vehicle", prop)
    mysqlDF.createOrReplaceTempView("zhjg_jsy_ticket_smz_curmonth")
    vehicleDF.createOrReplaceTempView(("zhjg_zy_vehicle"))

    import spark.sql
    //实际载客量：按照线路进行分组
    sql("SELECT XL,count(*) as p from zhjg_jsy_ticket_smz_curmonth where FCRQSJ like '%2019-04%' and SFLSXFC='N' and SCKCCPHM!=' ' GROUP BY XL")
      .createOrReplaceTempView("sjzkl")

    //线路上的车sjzkl
    sql("select SCKCCPHM,XL from zhjg_jsy_ticket_smz_curmonth  where FCRQSJ like '%2019-04%'")
      .filter("SFLSXFC='N'")
      .filter("SCKCCPHM!=' '")
      .groupBy("SCKCCPHM")
      .count()
      .createOrReplaceTempView("xlHasCar")

    //车辆核定座位数
    sql("select DISTINCT a.SCKCCPHM,a.XL,b.SEAT_COUNT,b.VEH_NO from zhjg_jsy_ticket_smz_curmonth as a,zhjg_zy_vehicle as b")
      .filter("a.SCKCCPHM = b.VEH_NO")
      .createOrReplaceTempView("hdzkl")

    //查找线路上面的车，以及车的座位数
    sql("select DISTINCT hdzkl.SCKCCPHM,hdzkl.XL,hdzkl.SEAT_COUNT from hdzkl,xlHasCar where xlHasCar.SCKCCPHM=hdzkl.SCKCCPHM")
      .createOrReplaceTempView("xlAndSeat")

    //车次
    sql("select SCKCCPHM,XL,FCRQSJ from zhjg_jsy_ticket_smz_curmonth where SFLSXFC='N' AND FCRQSJ like '%2019-04%' and SCKCCPHM!=' ' GROUP BY XL,SCKCCPHM,FCRQSJ")
      .createOrReplaceTempView("checi")

    sql("select SCKCCPHM,XL,FCRQSJ from checi")
      .groupBy("SCKCCPHM")
      .count()
      .createOrReplaceTempView("checi1")

    //核定载客量总数
    sql("select xlAndSeat.SCKCCPHM,xlAndSeat.XL,xlAndSeat.SEAT_COUNT,checi1.count,xlAndSeat.SEAT_COUNT*checi1.count as total from xlAndSeat,checi1 where xlAndSeat.SCKCCPHM=checi1.SCKCCPHM")
      .createOrReplaceTempView("hdzklzs")

    import spark.implicits._
    sql("select hdzklzs.XL,hdzklzs.total from hdzklzs")
      .groupBy("XL")
      .sum("total")
      .withColumn("total", $"sum(total)")
      .createTempView("alhdzks")
    //XL，实际载客量，核定载客量

    //loadFactors：计算载客率
    val loadFactors = sql("select sjzkl.XL,sjzkl.p,alhdzks.XL,alhdzks.total from sjzkl,alhdzks").filter("alhdzks.XL=sjzkl.XL").collect()

    for (i <- 0 to loadFactors.length - 1) {
      println(loadFactors(i)(0).toString + "载客率：" + loadFactors(i)(1).toString.toDouble / loadFactors(i)(3).toString.toDouble)
    }

  }
}
